
import java.io.*;
import java.util.*;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.conf.*;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.*;
import org.apache.hadoop.mapreduce.lib.input.*;
import org.apache.hadoop.mapreduce.lib.output.*;
import org.apache.hadoop.util.*;

public class HMMMapReduce extends Configured implements Tool {

   public static class Map
       extends Mapper<LongWritable, Text, Text, DoubleWritable> {

//     static enum Counters { INPUT_WORDS }

//     private final static DoubleWritable one = new DoubleWritable(1);
     private Text ProfileName = new Text();
     private Text InputArray = new Text();
     
     private HMMProfile HMM=new HMMProfile();
     private ReadHMMProfile ReadProfile;
     private ProteinSeq Sequence=new ProteinSeq();
     private ReadProteinSeq ReadProcess=new ReadProteinSeq();
     private Viterbi ViterbiProcess=new Viterbi();
     
 	 private boolean FileBegin=false;
 	 public LineNumberReader lnreader = null;
 	 String ProteinFileName;
 	 String BufferLine;
 	 boolean Flag_SeqNo=false;

//     private long numRecords = 0;
     private String inputFile;

     public void setup(Context context) {
       Configuration conf = context.getConfiguration();
       inputFile = conf.get("mapreduce.map.input.file");

       if (conf.getBoolean("HMMProfile.input.file", false)) {
         Path[] profileFiles = new Path[0];
         try {
           profileFiles = DistributedCache.getLocalCacheFiles(conf);
         } catch (IOException ioe) {
           System.err.println("Caught exception while getting cached files: "
               + StringUtils.stringifyException(ioe));
         }
         for (Path profileFile : profileFiles) {
     		 ReadProfile = new ReadHMMProfile(profileFile.toString());
         }
       }
     }

     public void map(LongWritable key, Text value, Context context)
         throws IOException, InterruptedException {

/*       if ((++numRecords % 100) == 0) {
         context.setStatus("Finished processing " + numRecords
             + " records " + "from the input file: " + inputFile);
       }*/
    	 
    	 
    	 context.setStatus("Profile File:" + inputFile);
    	 
    	 String line=value.toString();
    	 
 			try {
 				int Flag=ReadProcess.MapToProteinSeq(Sequence, line);
 				if(Flag==1|| Flag==2){
 					
 					while(ReadProfile.ReadNextProfile( HMM ))
 					{
 						Double score=ViterbiProcess.ViterbiAlgorithm(Sequence, HMM);
 						String SeqNo1=new String();
 						SeqNo1=Sequence.GetSeqNo();
 						InputArray.set(HMM.GetProfileName()+": "+Sequence.GetSeqNo()+": ");
 						context.write(InputArray, new DoubleWritable(score));
 					}
 					ReadProfile.Reset();
 					
 					/*InputArray.set(Sequence.GetSeqNo());
 					context.write(InputArray, new DoubleWritable(1));*/
 				}
 			} catch (IOException e) {
 				// TODO Auto-generated catch block
 				e.printStackTrace();
 			}
 	
     }
   }

   public static class Reduce
       extends Reducer<Text, DoubleWritable, Text, DoubleWritable> {
     public void reduce(Text key, Iterable<DoubleWritable> values,
         Context context) throws IOException, InterruptedException {

/*       int sum = 0;
       for (DoubleWritable val : values) {
         sum += val.get();
       }*/
    	 for (DoubleWritable val: values)
    		 context.write(key, val);
     }
   }

   public int run(String[] args) throws Exception {
     Job job = new Job(getConf());
     job.setJarByClass(HMMMapReduce.class);
     job.setJobName("HMMMapReduce");

     job.setOutputKeyClass(Text.class);
     job.setOutputValueClass(DoubleWritable.class);

     job.setMapperClass(Map.class);
     job.setCombinerClass(Reduce.class);
     job.setReducerClass(Reduce.class);

     job.setInputFormatClass(TextInputFormat.class);
     job.setOutputFormatClass(TextOutputFormat.class);

     List<String> other_args = new ArrayList<String>();
     for (int i=0; i < args.length; ++i) {
       if ("-profile".equals(args[i])) {
         DistributedCache.addCacheFile(new Path(args[++i]).toUri(),
             job.getConfiguration());
         job.getConfiguration().setBoolean("HMMProfile.input.file", true);
       } else {
         other_args.add(args[i]);
       }
     }

     FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
     FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));

     boolean success = job.waitForCompletion(true);
     return success ? 0 : 1;
   }

/*   public static void main(String[] args) throws Exception {
     int res = ToolRunner.run(new Configuration(), new HMMMapReduce(), args);
     System.exit(res);
   }*/
   
}